package com.sailing.xjpb.elasticsearch;

import com.google.common.collect.Lists;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.JestResult;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.cluster.Health;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
import org.apache.http.ssl.SSLContexts;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.net.ssl.SSLContext;
import java.io.*;
import java.security.KeyStore;
import java.util.List;
import java.util.Objects;

@Slf4j
@Component
@PropertySource(value = {"classpath:application-es.properties"})
public class JestClientConfigurationPool implements PooledObjectFactory<JestClient> {

    public JestClientConfigurationPool() {
        log.info("初始化:JestClientConfigurationPool");
    }

    private GenericObjectPool<JestClient> objectPool;
    private GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
    @Value("${es.cluster.clusterUrls:http://127.0.0.1:9200}")
    private String clusterUrls;
    @Value("${es.cluster.connect.pool.MaxIdle:500}")
    private int maxIdle;
    @Value("${es.cluster.connect.pool.MinIdle:1}")
    private int minIdle;
    @Value("${es.cluster.connect.pool.MaxTotal:500}")
    private int maxTotal;
    @Value("${es.cluster.connect.pool.MaxWaitMillis:1000}")
    private long maxWaitMillis;
    @Value("${es.cluster.connect.pool.MinEvictableIdleTimeMillis:20000}")
    private long minEvictableIdleTimeMillis;
    @Value("${es.cluster.connect.pool.TestWhileIdle:true}")
    private boolean testWhileIdle;
    @Value("${es.cluster.connect.pool.TestOnCreate:true}")
    private boolean testOnCreate;
    @Value("${es.cluster.connect.pool.TestOnBorrow:false}")
    private boolean testOnBorrow;
    @Value("${es.cluster.connect.pool.BlockWhenExhausted:false}")
    private boolean blockWhenExhausted;
    @Value("${es.cluster.connect.pool.TimeBetweenEvictionRunsMillis:100000}")
    private long timeBetweenEvictionRunsMillis;
    @Value("${es.client.connect.connTimeout:10000}")
    private int connTimeout;
    @Value("${es.client.connect.readTimeout:120000}")
    private int readTimeout;
    @Value("${es.certificates.p12.path}")
    private String esCertificatesP12Path;
    @Value("${es.certificates.p12.password}")
    private String esCertificatesP12Password;

    @PostConstruct
    public void init() {
        poolConfig.setMaxIdle(maxIdle);// 最大空闲资源
        poolConfig.setMinIdle(minIdle);// 最小空闲对象
        poolConfig.setMaxTotal(maxTotal);// 允许创建的对象个数

        poolConfig.setMaxWaitMillis(maxWaitMillis);// 对象耗尽后最多等待毫秒数。
        poolConfig.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);// 最小空闲时间
        poolConfig.setTestWhileIdle(testWhileIdle);

        poolConfig.setTestOnCreate(testOnCreate);// 当创建时验证
        poolConfig.setTestOnBorrow(testOnBorrow);// 获取时验证

        poolConfig.setBlockWhenExhausted(blockWhenExhausted);
        poolConfig.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);// 回收资源线程的执行周期

        objectPool = new GenericObjectPool<>(this, poolConfig);
    }

    public JestClient getConnection() throws Exception {
        return objectPool.borrowObject();
    }

    public void close(JestClient conn) {
        if (conn != null) {
            objectPool.returnObject(conn);
        }
    }

    private InputStream getFile() {
        File file = new File(esCertificatesP12Path);
        try {
            InputStream inputStream = new FileInputStream(file);
            return inputStream;
        } catch (FileNotFoundException e) {
            log.error("获取ES认证文件失败：{}", esCertificatesP12Path, e);
        }
        return null;
    }

    @Override
    public PooledObject<JestClient> makeObject() throws Exception {
        List<String> clusterList = Lists.newArrayList(clusterUrls.split(","));
        JestClientFactory factory = new JestClientFactory();
        HttpClientConfig.Builder configBuilder = new HttpClientConfig.Builder(clusterList) // 参数可以是集群，请先定义一个list
                // 集合，将节点url
                // 分别添加到list
                // .defaultCredentials(username,password) //如果使用了x-pack，就要添加用户名和密码
                .multiThreaded(true) // 多线程模式
                .connTimeout(connTimeout) // 连接超时
                .readTimeout(readTimeout);// 由于是基于http，所以超时时间必不可少，不然经常会遇到socket异常：read time out

        if (StringUtils.isNotBlank(esCertificatesP12Path)) {
            KeyStore truststore = KeyStore.getInstance("PKCS12");
            String keyStorePwd = Objects.toString(esCertificatesP12Password, "");
            truststore.load(getFile(), keyStorePwd.toCharArray());
            SSLContext sslcontext =
                    SSLContexts.custom().loadTrustMaterial(truststore, new TrustSelfSignedStrategy()).build();
            NoopHostnameVerifier hostnameVerifier = NoopHostnameVerifier.INSTANCE;
            SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslcontext, hostnameVerifier);
            configBuilder.sslSocketFactory(sslsf);
        }
        factory.setHttpClientConfig(configBuilder.build());
        log.debug("创建链接:{}", factory.hashCode());

        return new DefaultPooledObject<>(factory.getObject());
    }


    @Override
    public void destroyObject(PooledObject<JestClient> pooledObject) throws Exception {
        JestClient jestClient = pooledObject.getObject();
        if (jestClient != null) {
            try {
                jestClient.close();
                log.debug("关闭链接.");
            } catch (IOException e) {
                log.error("关闭连接异常", e);
            } finally {
                jestClient.close();
            }
        }
    }

    @Override
    public boolean validateObject(PooledObject<JestClient> pooledObject) {
        log.debug("连接池空闲{}个", objectPool.getNumIdle());
        log.debug("连接池活跃{}个", objectPool.getNumActive());
        log.debug("连接池等待{}个", objectPool.getNumWaiters());

        Health health = new Health.Builder().build();

        try {
            JestResult result = pooledObject.getObject().execute(health);
            if (result.isSucceeded()) {
                log.debug("集群名称:{}", result.getJsonObject().get("cluster_name").getAsString());
                log.debug("集群状态:{}", result.getJsonObject().get("status").getAsString());
            } else {
                log.warn("验证失败,检查链路.ES集群服务器.");
            }
        } catch (IOException e) {
            log.error("验证失败", e);
            return false;
        }
        return true;
    }

    @Override
    public void activateObject(PooledObject<JestClient> pooledObject) throws Exception {
        pooledObject.getObject();
        // log.info("JestClient Connection activate: " + connection.hashCode());
    }

    @Override
    public void passivateObject(PooledObject<JestClient> pooledObject) throws Exception {
        // JestClient connection = pooledObject.getObject();
        // log.info("JestClient Connection passivate: " + connection.hashCode());
    }
}
